package net.bwie.realtime.jtp.Rpfunction;

import net.bwie.realtime.jtp.bean.Order;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/05/28 11:01:29
 * @version: 1.0
 */
public class AbnormalOrderAlert extends PatternProcessFunction<Order, String> {
    private final int maxAbnormalCount;
    private final double largeAmountThreshold;
    private final double smallOrderRateThreshold;

    public AbnormalOrderAlert(int maxAbnormalCount,
                              double largeAmountThreshold,
                              double smallOrderRateThreshold) {
        this.maxAbnormalCount = maxAbnormalCount;
        this.largeAmountThreshold = largeAmountThreshold;
        this.smallOrderRateThreshold = smallOrderRateThreshold;
    }

    @Override
    public void processMatch(Map<String, List<Order>> match, Context ctx, Collector<String> out) throws Exception {
        Order firstOrder = match.get("first").get(0);
        Order secondOrder = match.get("second").get(0);

        // 构建报警信息
        String alertMsg = String.format(
                "【异常交易警报】\n" +
                        "时间: %s\n" +
                        "用户ID: %d\n" +
                        "首次订单金额: %.2f\n" +
                        "二次订单金额: %.2f\n" +
                        "规则: 短时间内连续大额交易(5分钟内两次超过%.2f)",
                new Date(ctx.timestamp()),
                firstOrder.getUserId(),
                firstOrder.getPayment(),
                secondOrder.getPayment(),
                largeAmountThreshold * 0.5
        );

        out.collect(alertMsg);
    }
}
